package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.bean.StockKlineBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;

/**
 * 个股分时窗口函数
 */
public class StockKlineWindowFunction extends RichWindowFunction<CleanBean, StockKlineBean, String, TimeWindow> {
    /**
     * 开发步骤：
     * 1.记录最新个股
     * 2.计算K线均价
     * 3.格式化日期
     * 4.封装输出数据
     */

    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<StockKlineBean> out) throws Exception {

        //1.记录最新个股
        CleanBean line = null; //缓存的是上一条数据
        for (CleanBean cleanBean : input) {
            if(line == null){
                line = cleanBean;
            }
            if(line.getEventTime()<cleanBean.getEventTime()){
                line = cleanBean;
            }
        }

//        //2.计算K线均价
//        //成交总额/成交总量
//        BigDecimal avgPrice = BigDecimal.valueOf(line.getTradeAmt()).divide(BigDecimal.valueOf(line.getTradeVolumn()), 2, RoundingMode.HALF_UP);

        // 3.格式化日期
        String tradeDate = DateUtil.longTimeToString(line.getEventTime(), Constant.format_yyyy_mm_dd);

        // 4.封装输出数据
        out.collect(new StockKlineBean(
            tradeDate,
                line.getSecCode(),
                line.getSecName(),
                line.getPreClosePrice(),
                line.getOpenPrice(),
                line.getHighPrice(),
                line.getLowPrice(),
                line.getTradePrice(),
                line.getTradeVolumn(),
                line.getTradeAmt(),
                line.getEventTime(),
                line.getSource(),
                1 //表示日K
        ));


    }
}
